package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T>{

    StatefulRedisConnection<String, String> redisAsyncConnection;
    AsyncConnection hBaseAsyncConnection;
    @Override
    public void open(Configuration parameters) throws Exception {
        redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
        hBaseAsyncConnection = HBaseUtil.getAsyncHBaseConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);
        HBaseUtil.closeAsyncHBaseConnection(hBaseAsyncConnection);
    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {
        CompletableFuture
                .<JSONObject>supplyAsync(
                new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        return RedisUtil.readDimAsync(redisAsyncConnection, getTableName(), getRowKey(t));
                    }
                }
        ).thenApplyAsync(
                new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimJsonObj) {
                        if (dimJsonObj == null){
                            // redis没有数据
                            dimJsonObj = HBaseUtil.readDimAsync(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, getTableName(), getRowKey(t), JSONObject.class);
                            if (dimJsonObj != null){
                                System.out.println("~~~从HBase中获取维度表" + getTableName() + "的" + getRowKey(t) + "数据~~~");
                                RedisUtil.writeDimAsync(redisAsyncConnection, getTableName(), getRowKey(t), dimJsonObj);
                            } else {
                                System.out.println("~~~没有找到维度表" + getTableName() + "的" + getRowKey(t) + "数据~~~");
                            }
                        } else {
                            System.out.println("~~~从Redis中获取维度表" + getTableName() + "的" + getRowKey(t) + "数据~~~");
                        }
                        return dimJsonObj;
                    }
                }
        ).thenAcceptAsync(
                new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimJsonObj) {
                        addDims(t, dimJsonObj);
                        resultFuture.complete(Collections.singleton(t));
                    }
                }
        );
    }
}
